package com.ibm.spatiotemp.utils;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
import java.util.HashSet;

import org.apache.log4j.Logger;

import com.ibm.spatiotemp.server.datamodel.gpsdata.TSPoint;
import com.ibm.spatiotemp.server.datasources.DataSource;

/**
 * Manage a thread pool to insert data into database continuously.
 * There is one DBThreadPool instance for one table type.
 *
 */
public class DBThreadPool{	
	/**
	 * Reference for log4j
	 */
	public final static Logger logger = Logger.getLogger(DBThreadPool.class);
	public final static TSPoint flushPoint = new TSPoint(null, 0, 0, 0);
	
	/**
	 * The flag to indicate to stop consumers
	 */
	boolean stopFlag = false;
	/**
	 * The queue of events. A event is a TSPoint.
	 */
	Queue[] _queues;
	/**
	 *  The collection of consumer threads. 
	 */
	Consumer[] _consumers;
	/**
	 * The router to select queue
	 */
	SimpleRouter dbRouter = null;
	ConnectionManager connManager = null;
	Connection firstConn = null;
	HashSet<String> idSet = null;
	FlushController flushController = null;
	/*
	 * For test
	 */
	long dbStartTime = 0;
	long[] _counters;

	/**
	 * Subclass of Thread. This nested class implements a consumer thread. One consumer is mapped to one queue.
	 */
	class Consumer extends Thread {
		/**
		 * The title of Consumer
		 */
		private String title;
		/**
		 * Connection for db insert
		 */
		private Connection conn;
		/**
		 * Queue for insert event
		 */
		private Queue queue = null;
		/**
		 * Object to insert data to DB by batch
		 */
		private TSPointToDB point2DB = null;
		/**
		 * for test
		 */
		private int index;

		
		public Consumer(String title, Connection conn, Queue queue, String sqlString, int index) throws SQLException {
			this.title = title;			
			this.conn = conn;	
			this.queue = queue;
			this.index = index;
			this.point2DB = new TSPointToDB(conn, sqlString, ConfigManager.BATCH_BUFFER_SIZE); 
		}
		

		public void run() {
			logger.info("Consumer[" + title + "] begins to insert data......");			
			while (!stopFlag) {
				try{					
					TSPoint event;
					// take an array
					Entry entry = queue.takeAll();
					// set up dbstarttime for test
					if (dbStartTime == 0) {
						dbStartTime = System.currentTimeMillis();
					}
					Entry unit;
					while (entry != null && !stopFlag) {
						unit = entry;
						entry = entry.getNext();
						event = (TSPoint)unit.getElement();
						unit.clear();
						consume(event);
						if (ConfigManager.TEST_MODE) {
							_counters[index] += 1;
						}
					}
				}catch(InterruptedException e){
					if (stopFlag) {
						logger.info("Consumer[" + title + "] is stopped.");	
						return;
					} else {
						logger.error("Consumer[" + title + "] got InterruptedException: " + e.getMessage());	
						continue;
					}
				} catch (SQLException e) {
					if (stopFlag) {
						logger.info("Consumer[" + title + "] is stopped.");	
					} else {
						logger.error("Consumer[" + title + "] got SQLException: (" + e.getMessage() + ", errorCode = " + e.getErrorCode() + ") and prepared to stop.");	
			    		// stop server			    		
						stopFlag = true;
					}
					break;
				}
			} // End while
			close();
			logger.info("Consumer[" + title + "] is stopped.");	
		} // End run
		
		/**
		 *  Processes an event (TSPoint): insert data into database using batch insert
		 *  @param event the event to be processed.
		 * @throws SQLException 
		 */
		public void consume(TSPoint event) throws SQLException{
			point2DB.addPoint(event);			
		}
		
		/**
		 * Close database connection and finish reporting
		 */
		public void close() {
			try {
				conn.close();
			} catch (SQLException e) {
			}
		}
		
	} // End Consumer
	
	
	/**
	 * Construct an instance of threadpool	 
	 * @throws SQLException 
	 * @throws RODBException 
	 */
	public DBThreadPool(DataSource ds, int numOfConsumers, int maxQueueLength) throws SQLException {
		// assert numOfWorkers > 0
		_queues = new Queue[numOfConsumers];
		_consumers = new Consumer[numOfConsumers];
		connManager = new ConnectionManager(ds);
		Connection conn = null;
		String sqlString = IfxInsertStringFormatter.insertPrepareString(ds.getDataschema());
		for (int i = 0; i < numOfConsumers; i++) {
			_queues[i] = new Queue(maxQueueLength);
			String title = "T_" + ds.getSrcname() + "_" + i;
			conn = connManager.newConnection();
			if (firstConn == null) {
				firstConn = conn;
			}
			_consumers[i] = new Consumer(title, conn, _queues[i], sqlString, i);
		}
		if (ConfigManager.TEST_MODE) {
			this._counters = new long[numOfConsumers];
			for (int i=0; i< numOfConsumers; i++) {
				_counters[i] = 0;
			}
		}
		this.dbRouter = new SimpleRouter();
		this.dbRouter.init(_queues);
		// init idSet
		idSet = connManager.initIDSet(firstConn);
		flushController = new FlushController(ConfigManager.FLUSH_INTERVAL, this);
	}
	
	public void startConsumers() {
		for (int i=0; i<_consumers.length; i++) {
			_consumers[i].start();	
		}
		flushController.start();
	}
	
	/** 
	  * Places an event on the queue. 
	  *
	  * @param event the event to be added.
	  */
	public synchronized void put(TSPoint event) {
		String modid = event.tsid;
		if (modid == null) {
			return;
		}
		// check modid
		if (!idSet.contains(modid)) {
			try {
				connManager.initPrimaryKey(firstConn, modid);
			} catch (SQLException e) {
				e.printStackTrace();
				logger.error("Fail to init primary key for modid=" + modid);
				return;
			}
			idSet.add(modid);
		}
		Queue queue = dbRouter.assignQueue(modid);
		if (queue == null) {
			// fail to assign queue
			logger.error("Fail to assign queue for modid=" + modid);
			return;
		}
		try {
			queue.put(event);
		} catch (InterruptedException e) {
			return;
		}	
	}
	
	/** 
	  * put flush event to all queues
	 * @throws InterruptedException 
	  */
	public synchronized void flush() {
		for (int i=0; i<_queues.length; i++) {
			try {
				_queues[i].put(flushPoint);
			} catch (InterruptedException e) {
			}
		}
	}
	
	/** 
	  * Places an event on the queue. 
	  *
	  * @param event the event to be added.
	  */
	public synchronized void put(String modid, Collection<TSPoint> elements) {
		Queue queue = dbRouter.assignQueue(modid);
		if (queue == null) {
			// fail to assign queue
			logger.error("Fail to assign queue for modid=" + modid);
			return;
		}
		// check modid
		if (!idSet.contains(modid)) {
			try {
				connManager.initPrimaryKey(firstConn, modid);
			} catch (SQLException e) {
				e.printStackTrace();
				logger.error("Fail to init primary key for modid=" + modid);
				return;
			}
			idSet.add(modid);
		}
		try {
			queue.putPoints(elements);
		} catch (InterruptedException e) {
			return;
		}	
	}
	

	/** 
	 * Finishes the use of the event queue and frees all resources. 
	 */
	public void close() {
		for (int i=0; i<_consumers.length; i++) {
			_consumers[i].close();
		}		
	}

	/**
	 * check if the queue is empty
	 * @return true if there are events in queue
	 */
	public boolean isEmpty() {
		boolean isEmpty = true;
		for (int i=0; i<_queues.length; i++) {
			if (!_queues[i].isEmpty()) {
				isEmpty = false;
				break;
			}
		}
		return isEmpty;
	}
	
	/**
	 * Check if all consumers are dead
	 * @return true if all consumers are dead
	 */
	public boolean isDead() {
		boolean dead = true;
		for (int i=0; i<_consumers.length; i++) {
			if (_consumers[i].isAlive()) {
				dead = false;
				break;
			}
		}
		return dead;
	}
	
	public boolean isStopFlag() {
		return stopFlag;
	}

	public synchronized void setStopFlag(boolean stopFlag) {
		this.stopFlag = stopFlag;
	}
	
	public void stopConsumers() {
		flush();
		stopFlushController();
		try {
			Thread.sleep(1000);					
		} catch (InterruptedException e) {}
		synchronized(this) {
			this.stopFlag = true;
		}		
		for (int i=0; i<_consumers.length; i++) {
			_consumers[i].interrupt();
		}
		// make sure Consumer is dead
		boolean notFinish = true;
		while (notFinish) {
			notFinish = false;
			boolean alive;
			for (int i=0; i<_consumers.length; i++) {
				alive = _consumers[i].isAlive();
				if (alive) {
					notFinish = true;
					break;
				}				
			}
			logger.info("DBThreadPool waits consumers to finish writing data into database and sleep 1 second.");
			try {
				Thread.sleep(1000);					
			} catch (InterruptedException e) {}
		}
	}
	
	public void stopFlushController() {
		flushController.setStopFlag(true);
		flushController.interrupt();
	}
	
	
	public int getNumberInQueue() {
		int number = 0;
		for (int i=0; i<_queues.length; i++) {
			number += _queues[i].size();
		}
		return number;
	}
	
	public long getTotal() {
		long total = 0;
		for (int i=0; i<_counters.length; i++) {
			total += _counters[i];
		}
		return total;
	}

	public long getDbStartTime() {
		return dbStartTime;
	}

	public Queue[] getQueues() {
		return _queues;
	}
	
	public Connection newConnection() throws SQLException {
		return connManager.newConnection();
	}
	
	
	
} // End DBThreadPool